package com.jiuli.kafkaleaning.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.jms.AcknowledgeMode;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
@Slf4j
public class KafkaConsumer {

//    @KafkaListener(topics = {"test-topic-1"})
//    public void listenerRecord(ConsumerRecord<String, String>
//                                             record, Acknowledgment ack) {
//        try {
//            log.info("Received message: topic - {}, partition - {}, " +
//                            "offset - {}, key - {}, value - {}",
//                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
//        }catch (Exception e) {
//            e.printStackTrace();
//            throw new RuntimeException("消息消费异常");
//        }
//        ack.acknowledge();
//    }

    @KafkaListener(topics = {"test-topic-1"}, groupId = "csbjcsbj3")
    public void listenBatch(List<Message<?>> list, Acknowledgment ack, Consumer<?, ?> consumer) {
        try {
            log.info("Received message: {}", list);
            log.info("Received consumer: {}", consumer.assignment());
        }catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("消息消费异常");
        }
        ack.acknowledge();
    }
}
